using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{
    /// <summary>
    ///  This applies a Template Method pattern, much like Spring's JdbcTemplate, to
    ///  Cassandra. The ColumnFamilyTemplate instance maintains many of the fields in
    ///  common between various query/update operations so that they do not need to be
    ///  constantly passed for every operation on the column family. These include the
    ///  keyspace, column family name, key serializer, and the column name serializer
    ///  (for standard column name or the super column name).
    ///  
    ///  the key and column name type. It defers the generic types for super column
    ///  The Java generic types of the ColumnFamilyTemplate class itself are limited to
    ///  child types to the individual update/query operation.
    ///  
    ///  Note: this class is abstract and doesn't implement update, subclasses have to
    ///  provide the logic for executing queries.
    ///  </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
	public abstract class ColumnFamilyTemplate<K,N> : AbstractColumnFamilyTemplate<K,N>
	{
        public ColumnFamilyTemplate(Keyspace keyspace, String columnFamily,
            Serializer<K> keySerializer, Serializer<N> topSerializer) : base(keyspace, columnFamily, keySerializer, topSerializer)
        {
        }
        public new ColumnFamilyTemplate<K, N> SetBatched(bool batched)
        {
            base.Batched = batched;
            return this;
        }
        public ColumnFamilyUpdater<K, N> CreateUpdater()
        {
            ColumnFamilyUpdater<K, N> updater = new ColumnFamilyUpdater<K, N>(this, ColumnFactory);
            updater.Clock = Keyspace.CreateClock();
            updater.ConsistencyLevelPolicy = Keyspace.ConsistencyLevelPolicy;
            return updater;
        }
        public ColumnFamilyUpdater<K, N> CreateUpdater(K key)
        {
            ColumnFamilyUpdater<K, N> updater = new ColumnFamilyUpdater<K, N>(this, ColumnFactory);
            updater.Clock = Keyspace.CreateClock();
            updater.ConsistencyLevelPolicy = Keyspace.ConsistencyLevelPolicy;
            updater.AddKey(key);
            return updater;
        }
        public ColumnFamilyUpdater<K, N> CreateUpdater(K key, Mutator<K> mutator)
        {
            ColumnFamilyUpdater<K, N> updater = new ColumnFamilyUpdater<K, N>(this, ColumnFactory, mutator);
            updater.Clock = Keyspace.CreateClock();
            updater.ConsistencyLevelPolicy = Keyspace.ConsistencyLevelPolicy;
            updater.AddKey(key);
            return updater;
        }
        public void Update(ColumnFamilyUpdater<K, N> updater)
        {
            updater.Update();
            ExecuteIfNotBatched(updater);
        }
        public bool IsColumnsExist(K key)
        {
            return CountColumns(key) > 0;
        }
        public int CountColumns(K key)
        {
            return CountColumns(key, (N)ALL_COLUMNS_START, (N)ALL_COLUMNS_END,
                ALL_COLUMNS_COUNT);
        }
        public int CountColumns(K key, N start, N end, int max)
        {
            CountQuery<K, N> query = HFactory.CreateCountQuery(Keyspace, KeySerializer,
                TopSerializer);
            query.SetKey(key);
            query.SetColumnFamily(ColumnFamily);
            query.SetRange(start, end, max);
            return query.Execute().Get();
        }
        public ColumnFamilyResult<K, N> QueryColumns(K key)
        {
            return DoExecuteSlice(key, ActiveSlicePredicate);
        }
        public ColumnFamilyResult<K, N> QueryColumns(IEnumerable<K> keys)
        {
            return DoExecuteMultigetSlice(keys, ActiveSlicePredicate);
        }
        public T QueryColumns<T>(K key, ColumnFamilyRowMapper<K, N, T> mapper)
        {
            return QueryColumns(key, ActiveSlicePredicate, mapper);
        }
        public T QueryColumns<T>(K key, HSlicePredicate<N> predicate,
            ColumnFamilyRowMapper<K, N, T> mapper)
        {
            return DoExecuteSlice(key, predicate, mapper);
        }
        public T QueryColumns<T>(K key, List<N> columns,
            ColumnFamilyRowMapper<K, N, T> mapper)
        {
            HSlicePredicate<N> predicate = new HSlicePredicate<N>(TopSerializer);
            predicate.SetColumnNames(columns);
            return DoExecuteSlice(key, predicate, mapper);
        }
        public ColumnFamilyResult<K, N> QueryColumns(K key, List<N> columns)
        {
            HSlicePredicate<N> predicate = new HSlicePredicate<N>(TopSerializer);
            predicate.SetColumnNames(columns);
            return DoExecuteSlice(key, predicate);
        }
        public ColumnFamilyResult<K, N> QueryColumns(IEnumerable<K> keys, List<N> columns)
        {
            HSlicePredicate<N> predicate = new HSlicePredicate<N>(TopSerializer);
            predicate.SetColumnNames(columns);
            return DoExecuteMultigetSlice(keys, predicate);
        }
        public ColumnFamilyResult<K, N> QueryColumns(K key, HSlicePredicate<N> predicate)
        {
            return DoExecuteSlice(key, predicate);
        }
        public MappedColumnFamilyResult<K,N,V> QueryColumns<V>(IEnumerable<K> keys,
            ColumnFamilyRowMapper<K, N, V> mapper)
        {
            return DoExecuteMultigetSlice(keys, ActiveSlicePredicate, mapper);
        }
        public MappedColumnFamilyResult<K,N,V> QueryColumns<V>(IEnumerable<K> keys,
            HSlicePredicate<N> predicate, ColumnFamilyRowMapper<K, N, V> mapper)
        {
            return DoExecuteMultigetSlice(keys, predicate, mapper);
        }
        public MappedColumnFamilyResult<K, N, V> QueryColumns<V>(IEnumerable<K> keys,
            List<N> columns, ColumnFamilyRowMapper<K, N, V> mapper)
        {
            HSlicePredicate<N> predicate = new HSlicePredicate<N>(TopSerializer);
            predicate.SetColumnNames(columns);
            return DoExecuteMultigetSlice(keys, predicate, mapper);
        }
        public ColumnFamilyResult<K,N> QueryColumns<V>(IndexedSlicesPredicate<K,N,V> predicate)
        {
            return DoExecuteIndexedSlices(predicate);
        }
        public ColumnFamilyResult<K,N> QueryColumns<V>(IndexedSlicesPredicate<K,N,V> predicate, HSlicePredicate<N> slicePredicate)
        {
            return DoExecuteIndexedSlices(predicate, slicePredicate);
        }
        public ColumnFamilyResult<K,N> QueryColumns<V>(IndexedSlicesPredicate<K,N,V> predicate, List<N> columns)
        {
            HSlicePredicate<N> slicePredicate = new HSlicePredicate<N>(TopSerializer);
            slicePredicate.SetColumnNames(columns);
            return DoExecuteIndexedSlices(predicate, slicePredicate);
        }
        public MappedColumnFamilyResult<K,N,R> QueryColumns<R,V>(IndexedSlicesPredicate<K,N,V> predicate, ColumnFamilyRowMapper<K, N, R> mapper)
        {
            return DoExecuteIndexedSlices(predicate, mapper);
        }
        public MappedColumnFamilyResult<K,N,R> QueryColumns<R,V>(IndexedSlicesPredicate<K,N,V> predicate, 
            HSlicePredicate<N> slicePredicate, ColumnFamilyRowMapper<K, N, R> mapper)
        {
            return DoExecuteIndexedSlices(predicate, slicePredicate, mapper);
        }
        public MappedColumnFamilyResult<K, N, R> QueryColumns<R, V>(IndexedSlicesPredicate<K, N, V> predicate, 
            List<N> columns, ColumnFamilyRowMapper<K, N, R> mapper)
        {
            HSlicePredicate<N> slicePredicate = new HSlicePredicate<N>(TopSerializer);
            slicePredicate.SetColumnNames(columns);
            return DoExecuteIndexedSlices(predicate, slicePredicate, mapper);
        }
        public HColumn<N, V> QuerySingleColumn<V>(K key, N columnName,
            Type valueClass)
        {
            return QuerySingleColumn(key, columnName, (Serializer<V>) SerializerTypeInferer.GetSerializer<V>(valueClass));
        }
        public HColumn<N, V> QuerySingleColumn<V>(K key, N columnName,
            Serializer<V> valueSerializer)
        {
            ColumnQuery<K, N, V> query = HFactory.CreateColumnQuery(Keyspace, KeySerializer, TopSerializer, valueSerializer);
            query.SetColumnFamily(ColumnFamily);
            query.SetKey(key);
            query.SetName(columnName);
            QueryResult<HColumn<N, V>> result = query.Execute();
            return result != null ? result.Get() : null;
        }

        #region Delegated methods
        public abstract T DoExecuteSlice<T>(K key, HSlicePredicate<N> predicate, ColumnFamilyRowMapper<K, N, T> mapper);
        public abstract ColumnFamilyResult<K, N> DoExecuteSlice(K key, HSlicePredicate<N> workingSlicePredicate);
        public abstract ColumnFamilyResult<K, N> DoExecuteMultigetSlice(IEnumerable<K> keys, HSlicePredicate<N> workingSlicePredicate);
        public abstract MappedColumnFamilyResult<K, N, V> DoExecuteMultigetSlice<V>(IEnumerable<K> keys, HSlicePredicate<N> workingSlicePredicate, ColumnFamilyRowMapper<K, N, V> mapper);
        public abstract ColumnFamilyResult<K, N> DoExecuteIndexedSlices<V>(IndexedSlicesPredicate<K, N, V> predicate);
        public abstract ColumnFamilyResult<K, N> DoExecuteIndexedSlices<V>(IndexedSlicesPredicate<K, N, V> predicate, HSlicePredicate<N> slicePredicate);
        public abstract MappedColumnFamilyResult<K, N, R> DoExecuteIndexedSlices<R,V>(IndexedSlicesPredicate<K,N,V> predicate, ColumnFamilyRowMapper<K, N, R> mapper);
        public abstract MappedColumnFamilyResult<K, N, R> DoExecuteIndexedSlices<R,V>(IndexedSlicesPredicate<K,N,V> predicate, HSlicePredicate<N> slicePredicate, ColumnFamilyRowMapper<K, N, R> mapper);
        #endregion
    }
}